package com.shujia.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo02UpdateStateByKey {
  def main(args: Array[String]): Unit = {
    // 统计每个单词的数据 需要维护每一次的计算结果
    val conf = new SparkConf()
    conf.setAppName("Demo02UpdateStateByKey")
    conf.setMaster("local[2]")
    val sc = new SparkContext(conf)

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    val lineDS: DStream[String] = ssc.socketTextStream("master", 8888)

    List

    /**
     *
     * @param seq :相当于是每个批次(每5s)中的数据
     * @param opt :相当于上一个批次运行的结果
     * @return 最终返回当前批次的运行结果
     */
    def updateStateByKeyFunc(seq: Seq[Int], opt: Option[Int]): Option[Int] = {
      Some(seq.sum + opt.getOrElse(0)) // 统计当前批次的单词数量 并加上上一批次计算的数量 最终以Some的形式返回
    }

    // 使用有状态算子需要先指定checkpoint的地址
    ssc.checkpoint("Spark/data/ssc/checkpoint")


    // 如果需要维护状态则需要使用有状态算子
    lineDS
      .flatMap(_.split(","))
      .map(word => (word, 1))
      //      .updateStateByKey(updateStateByKeyFunc) // 定义函数
      .updateStateByKey((seq: Seq[Int], opt: Option[Int]) => {
        Some(seq.sum + opt.getOrElse(0))
      }) // 使用匿名函数
      .print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
